#include <sdfs/protobuf/Codec.h>

#include <muduo/base/Logging.h>
#include <muduo/net/Endian.h>
#include <zlib.h>


using namespace sdfs;
using namespace sdfs::protobuf;
using namespace muduo::net;
using namespace muduo;

void ProtobufCodec::onMessage(const muduo::net::TcpConnectionPtr& conn,
			muduo::net::Buffer* buf,
			muduo::Timestamp recvTime)
{
	LOG_TRACE << "work";
	while(buf->readableBytes() >= kMinMessageLen + kHeaderLen)
	{
		int32_t len = buf->peekInt32();
		if(len > kMaxMessageLen || len < kMinMessageLen + kHeaderLen)
		{
			LOG_TRACE << "error";
			errorCallback_(conn, buf, recvTime, kInvalidLength);
			break;
		}
		else if(buf->readableBytes() >= implicit_cast<size_t>(len + kHeaderLen))
		{
			ErrorCode errorCode = kNoError;
			MessagePtr message = parse(buf->peek()+kHeaderLen, len,  &errorCode);
			if(errorCode == kNoError)
			{
				LOG_TRACE << "call messageCallback";
			 	messageCallback_(conn, message, recvTime);
				buf->retrieve(kHeaderLen + len);
			}
			else
			{
				LOG_TRACE << "error";
				errorCallback_(conn, buf, recvTime, errorCode);
				break;
			}
			
		}
		else
		{
			break;
		}
	}	
}

void ProtobufCodec::Send(const muduo::net::TcpConnectionPtr& conn, 
			const google::protobuf::Message& message)
{
	muduo::net::Buffer buffer;
	fillEmptyBuffer( &buffer,  message);
	conn->send(&buffer);
}



void ProtobufCodec::fillEmptyBuffer(muduo::net::Buffer* buf, 
			const google::protobuf::Message& message)
{
	assert(buf->readableBytes() == 0);
	
	const std::string& msgName = message.GetTypeName();
	int32_t nameLen =static_cast<int32_t> (msgName.size()+1);

	buf->appendInt32(nameLen);
	buf->append(msgName.c_str(), nameLen);

	int byteSize = message.ByteSize();
	buf->ensureWritableBytes(byteSize);

	uint8_t* start = reinterpret_cast<uint8_t*>(buf->beginWrite());
	uint8_t* end = message.SerializeWithCachedSizesToArray(start);

	if(end - start != byteSize)
		LOG_ERROR << "message has broken";

	buf->hasWritten(byteSize);
	int32_t adler = static_cast<int32_t>(::adler32(0, Z_NULL, 0));
	int32_t checkSum = static_cast<int32_t>(
		::adler32(adler, 
			reinterpret_cast<const Bytef*>(buf->peek()),
			static_cast<int>(buf->readableBytes())));
	buf->appendInt32(checkSum);
	int32_t len = static_cast<int32_t>(buf->readableBytes());

	buf->prependInt32(len);
}

namespace
{
	const string kNoErrorStr = "NoError";
	const string kInvalidLengthStr = "InvalidLength";
	const string kCheckSumErrorStr = "CheckSumError";
	const string kInvalidNameLenStr = "InvalidNameLen";
	const string kUnknownMessageTypeStr = "UnknownMessageType";
	const string kParseErrorStr = "ParseError";
	const string kUnknownErrorStr = "UnknownError";
}


google::protobuf::Message* ProtobufCodec::createMessage(const std::string& type_name)
{
	google::protobuf::Message* message = NULL;
	const google::protobuf::Descriptor* descriptor = 
		google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(type_name);
	if(descriptor)
	{
		const google::protobuf::Message* protobuf = 
			google::protobuf::MessageFactory::generated_factory()->GetPrototype(descriptor);
		if(protobuf)
		{
			message = protobuf->New();
		}
	}
	return message;
}

int32_t asInt32(const char* buf)
{
	int32_t be32 = 0;
	::memcpy(&be32, buf, sizeof(int32_t));
	return sockets::networkToHost32(be32);
}

MessagePtr ProtobufCodec::parse(const char* buf, int len, ProtobufCodec::ErrorCode* error)
{
	MessagePtr message;

	int32_t expectedCheckSum = asInt32(buf+len-sizeof(int32_t));
	int32_t adler = static_cast<int32_t>(::adler32(0, Z_NULL, 0));
	int32_t checkSum = static_cast<int32_t>(
		::adler32(adler,
			reinterpret_cast<const Bytef*>(buf),
			static_cast<int>(len - kHeaderLen)));
	if(checkSum == expectedCheckSum)
	{
		int32_t nameLen = asInt32(buf);
		if(nameLen > 1 && nameLen <= len - 2*kHeaderLen)
		{
			std::string typeName(buf+kHeaderLen,buf+kHeaderLen+nameLen-1);
			message.reset(createMessage(typeName));
			if(message)
			{
				const char* data = buf + kHeaderLen + nameLen;
				int32_t dataLen = len - kHeaderLen*2 - nameLen;
				if(message->ParseFromArray(data, dataLen))
				{
					*error = kNoError;
				}
				else
				{
					*error = kParseError;
				}
			}
			else
			{
				*error = kUnknownMessageType;
			}
		}
		else
			*error = kInvalidLength;
	}
	else
		*error = kCheckSumError;
	return message;
}

void ProtobufCodec::defaultErrorCallback(const muduo::net::TcpConnectionPtr& conn,
				muduo::net::Buffer* buf,
				muduo::Timestamp,
				ErrorCode code)
{
	LOG_ERROR << "ProtobufCodec::defaultErrorCallback - "<< errorCodeToString(code);
	if(conn && conn->connected())
	{
		conn->shutdown();
	}
}

const string& ProtobufCodec::errorCodeToString(ProtobufCodec::ErrorCode errorCode)
{
  switch (errorCode)
  {
   case kNoError:
     return kNoErrorStr;
   case kInvalidLength:
     return kInvalidLengthStr;
   case kCheckSumError:
     return kCheckSumErrorStr;
   case kInvalidNameLen:
     return kInvalidNameLenStr;
   case kUnknownMessageType:
     return kUnknownMessageTypeStr;
   case kParseError:
     return kParseErrorStr;
   default:
     return kUnknownErrorStr;
  }
}
